package com.example.datastreamdemo

import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object MySourceParlleFunctionDemo {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val function = new MyParalleSourceFunction()
    val res: DataStream[Long] = senv.addSource(function).setParallelism(3)
    res.print()


    senv.execute()

//    Thread.sleep(5)
//
//    function.cancel()

  }
}

class MyParalleSourceFunction extends ParallelSourceFunction[Long] {
  var ele:Long = 0
  var isRunning = true

  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
    while(isRunning) {
      ele += 1;
      ctx.collect(ele)
      Thread.sleep(1000)
    }
  }
}
